package com.demo;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.descriptors.ConnectorDescriptor;
import org.apache.flink.table.descriptors.FormatDescriptor;
import org.apache.flink.table.descriptors.Json;
import org.apache.flink.table.descriptors.Kafka;
import org.apache.flink.table.descriptors.Rowtime;
import org.apache.flink.table.descriptors.Schema;
import org.apache.flink.types.Row;

public class Main {

  public static void main(String[] args) throws Exception {
    EnvironmentSettings envSettings = EnvironmentSettings.newInstance()
        .useBlinkPlanner()
        .inStreamingMode()
        .build();
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, envSettings);

    ConnectorDescriptor connectorDescriptor = new Kafka()
        .version("0.10")
        .topic("flink-demo")
        .startFromLatest()
        .property("group.id", "test1")
        .property("zookeeper.connect", "10.24.67.106:2181")
        .property("bootstrap.servers", "10.24.67.108:9092");

    Schema schema = new Schema()
        .field("user_id", DataTypes.STRING())
        .field("last_value", DataTypes.DECIMAL(10, 2))
        .field("current_value", DataTypes.DECIMAL(10, 2))
        .rowtime(
            new Rowtime()
                .timestampsFromSource()
                .watermarksPeriodicBounded(6000)
        );

    FormatDescriptor formatDescriptor =
        new Json()
            .failOnMissingField(true)
//            .jsonSchema(
//                "{" +
//                    "  type: 'object'," +
//                    "  properties: {" +
//                    "    user_id: {" +
//                    "      type: 'string'" +
//                    "    }," +
//                    "    last_value: {" +
//                    "      type: 'number'" +
//                    "    }, " +
//                    "    current_value: {" +
//                    "      type: 'number'" +
//                    "    }, " +
//                    "    proctime: {" +
//                    "      type: 'string'," +
//                    "      format: 'date-time'" +
//                    "    }" +
//                    "  }" +
//                    "}"
//            )
            .deriveSchema();

    tableEnv
        .connect(connectorDescriptor)
        .withFormat(formatDescriptor)
        .withSchema(schema)
        .inAppendMode()
        .createTemporaryTable("speed_table");

    String sql = "select `current_value` - `last_value` as gap from speed_table";

    Table table = tableEnv.sqlQuery(sql);
    table.printSchema();

    tableEnv.toAppendStream(table, Row.class).print();

    tableEnv.execute("Test");
  }

}

// {"user_id":"1","send_time":"2019-12-31T17:34:01.000Z","current_value":123.9,"last_value":23.5}
// {"user_id":"1","proctime":"2019-12-31T17:34:01.000Z","current_value":123,"last_value":23}
// kafka-console-producer --broker-list 10.24.67.108:9092 --topic flink-demo
// kafka-topics --create --zookeeper 10.24.67.191:2181/kafka --replication-factor 1 --partitions 1 --topic yc-test
// kafka-topics --list --zookeeper 10.24.67.106:2181
// kafka-topics --delete --topic yc-test --zookeeper 10.24.67.191:2181/kafka